package practice.test3.studentRecord.plan2;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;

/**
 * 本题的另一种解题思路
 *
 * map阶段取出用于区分语数外的标志
 * reduce对不同的科目进行累加和求平均，以及调整输出格式
 */
public class OtherMethod extends Configured implements Tool {

    private static ScoreRecordParser parser=new ScoreRecordParser();

    static class CalculateScoreMapper extends Mapper<LongWritable,Text,Text,Text> {
        private Text key=new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException{
            if(parser.parse(value)){
                this.key.set(parser.getName());
                context.write(this.key,value);
            }
        }
    }

    static class CalculateScoreReducer extends Reducer<Text,Text,Text,Text> {
        private Text value=new Text();

        @Override
        protected void reduce(Text key,Iterable<Text> values,Context context) throws IOException, InterruptedException{
            StringBuffer buffer=new StringBuffer();
            double sum=0;
            for(Text text:values){
                if(parser.parse(text)){
                    String id=parser.getId();
                    String score=parser.getScore();
                    switch(id){
                        case "a":{
                            buffer.append("语文："+score+"\t");
                            break;
                        }
                        case "b":{
                            buffer.append("英语："+score+"\t");
                            break;
                        }
                        case "c":{
                            buffer.append("数学："+score+"\t");
                            break;
                        }
                    }
                    sum+=Double.parseDouble(score);
                }
                System.out.println("======================================");
                System.out.println(text);
            }
            buffer.append("总分："+sum+"\t平均分："+(sum/3));
            this.value.set(buffer.toString());
            context.write(key,this.value);
        }
    }

    @Override
    public int run(String[] args) throws Exception {

        Configuration conf=getConf();
//        Path input=new Path(conf.get("input"));
//        Path output=new Path(conf.get("output"));

        Job job=Job.getInstance(conf,this.getClass().getSimpleName());
        job.setJarByClass(this.getClass());

        job.setMapperClass(CalculateScoreMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(CalculateScoreReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        TextInputFormat.addInputPath(job,new Path("F:\\test\\StudentInput"));
        TextOutputFormat.setOutputPath(job,new Path("F:\\test\\StudentOutput"));

        return job.waitForCompletion(true)?0:1;
    }

    public static void main(String[] args) throws Exception {

//        args = new String[]{"F:\\test\\input", "F:\\test\\output"};
        System.exit(ToolRunner.run(new OtherMethod(),args));
    }
}
